package edu.hawaii.ics.pjdb.worker;

import java.util.concurrent.Semaphore;

import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TSimpleServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.hawaii.ics.pjdb.types.RemoteAddress;
import edu.hawaii.ics.pjdb.types.WorkerCommand;
import edu.hawaii.ics.pjdb.types.WorkerService;
import edu.hawaii.ics.pjdb.util.ThriftHelper;

public class WorkerHelper {
    private static final Logger logger = LoggerFactory
            .getLogger(WorkerHelper.class);

    /**
     * Creates a processor that uses the local connection to the database.
     * 
     * @param db
     * @return
     */
    public static WorkerService.Processor<WorkerService.Iface> createProcessor() {
        return new WorkerService.Processor<WorkerService.Iface>(
                new WorkerCommandHandler());
    }

    /**
     * Creates a worker that listens for commands on the specified port and with
     * the specified connection to a location database.
     * 
     * Note that you must call {@link TServer#serve()} to start listening for
     * commands!
     * 
     * @param port
     * @return
     * @throws TTransportException
     */
    public static TServer createWorker(int port) throws TTransportException {
        TServer server = new THsHaServer(new THsHaServer.Args(
                new TNonblockingServerSocket(port))
                .processor(createProcessor())
                .transportFactory(ThriftHelper.getTransportFactory())
                .protocolFactory(ThriftHelper.getProtocolFactory()));

        return server;
    }

    /**
     * Creates a {@link Runnable} that starts a worker in {@link Runnable#run()}
     * .
     * 
     * @param port
     * @param localDb
     * @return
     * @throws TTransportException
     */
    public static Runnable createdThreadedWorker(final int port,
            final String localDbPath) {
        return new Runnable() {
            @Override
            public void run() {
                try {
                    TServer server = new TSimpleServer(new TServer.Args(
                            new TServerSocket(port))
                            .processor(createProcessor())
                            .transportFactory(
                                    ThriftHelper.getTransportFactory())
                            .protocolFactory(ThriftHelper.getProtocolFactory()));
                    server.serve();
                } catch (TTransportException e) {
                    throw new RuntimeException(e);
                }
            }
        };
    }

    /**
     * Sends a command to a worker. Note that this does not do anything to setup
     * the results listeners.
     * 
     * @param command
     * @throws TException
     */
    public static void sendCommand(WorkerCommand command, Semaphore lock)
            throws TException {
        RemoteAddress workerAddr = command.getWorkerAddress();

        // Create client that sends command to worker.
        TSocket socket = new TSocket(workerAddr.getAddress(),
                workerAddr.getPort());
        TTransport transport = ThriftHelper.getTransportFactory().getTransport(
                socket);
        TProtocol protocol = ThriftHelper.getProtocolFactory().getProtocol(
                transport);

        // Send command and clean up
        transport.open();
        WorkerService.Client client = new WorkerService.Client(protocol);
        client.executeCommand(command);
        transport.close();

        logger.debug("Done with command -- {}", command.toString());

        // Release a permit to inform people who care that this task is done
        lock.release();
    }
}
